/**
 *    Copyright (c) [2022] [中和农信]
 *    rabbit-mq-plus is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 *
 */
package com.cfpamf.athean.rabbit.mq.plus.producer.core.impl;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.Resource;

import org.springframework.stereotype.Service;

import com.cfpamf.athean.rabbit.mq.plus.common.constant.BusinessConstants;
import com.cfpamf.athean.rabbit.mq.plus.common.enums.ProcessStatusEnum;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusEventTaskDTO;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusMessageDTO;
import com.cfpamf.athean.rabbit.mq.plus.producer.core.schedule.RabbitMqPlusProducerMessageRetryJob;
import com.cfpamf.athean.rabbit.mq.plus.service.RabbitMqPlusEventTaskService;
import com.cfpamf.athean.rabbit.mq.plus.service.RabbitMqPlusProducerService;
import com.cfpamf.athena.utils.DateUtils;
import com.cfpamf.athena.utils.JSONUtils;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

/**
 * 类DefaultRabbitMqPlusProducerMessageRetryJob.java的实现描述：
 *
 * @author Andy（陈洪桥）2022年12月10 00:00
 */
@Slf4j
@Service("rabbitMqPlusProducerMessageRetryJob")
public class DefaultRabbitMqPlusProducerMessageRetryJob implements RabbitMqPlusProducerMessageRetryJob {

    volatile Lock lock = new ReentrantLock();

    @Resource
    private RabbitMqPlusEventTaskService rabbitMqPlusEventTaskService;

    @Resource
    private RabbitMqPlusProducerService rabbitMqPlusProducerService;

    @Override
    public void runAutoRetryCheck() {
        try {
            if (!lock.tryLock(BusinessConstants.SIXTY, TimeUnit.SECONDS)) {// 重入锁，等待60s，不要同时多节点并发。
                if (log.isInfoEnabled()) {
                    log.info("[rabbitMqPlus]启用rabbitMq生产消息发送状态检查，执行时间：{} 上一批job还在处理中，拒绝执行。",
                        DateUtils.format(DateUtils.nowTime()));
                }
                return;
            }
            if (log.isInfoEnabled()) {
                log.info("[rabbitMqPlus]启用rabbitMq生产消息发送状态检查，执行时间：{}", DateUtils.format(DateUtils.nowTime()));
            }
            List<Integer> status = Lists.newArrayList();
            status.add(ProcessStatusEnum.TODO.getCode());
            status.add(ProcessStatusEnum.FAIL.getCode());
            List<RabbitMqPlusEventTaskDTO> data = rabbitMqPlusEventTaskService.queryTodoDataByStatus(status);
            for (RabbitMqPlusEventTaskDTO eventTaskDTO : data) {
                // 检查TODO消息发送状态为待处理的消息，&&retryCount < maxRetryCount
                if (ProcessStatusEnum.isNotDone(eventTaskDTO.getStatusCode())
                    && eventTaskDTO.getRetryCount() <= eventTaskDTO.getMaxRetryCount()) {
                    RabbitMqPlusMessageDTO rabbitMqPlusMessageDTO =
                        JSONUtils.jsonToBean(eventTaskDTO.getMessageBody(), RabbitMqPlusMessageDTO.class);
                    // 消息发送失败，补偿重新发送消息，更新DB重试次数和状态
                    rabbitMqPlusProducerService.retrySendMessage(rabbitMqPlusMessageDTO, eventTaskDTO.getRetryCount());
                }
            }
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("[rabbitMqPlus]启用rabbitMq生产消息发送状态检查，异常错误信息：{} ", e);
            }
        } finally {
            lock.unlock();
        }
    }

}
